#ifndef THREAD_MANAGER_H_
#define THREAD_MANAGER_H_

#include "ace/Task.h"
#include "ace/OS_NS_string.h"
#include "ace/OS_NS_time.h"
#include "ace/Containers.h"
#include "ace/Synch.h"
#include "ace/SString.h"
#include "ace/Method_Request.h"
#include "ace/Future.h"
#include "ace/Activation_Queue.h"
#include "ace/Condition_T.h"
#include "sys/types.h"


#include "ThreadWorker.h"

template<class SubWorkerClass> class IWorker;

template<class T>
class IManager
{
public:
	virtual ~IManager(void) { }

	virtual int return_to_work (IWorker<T> *worker) = 0;
};


template<class SubWorkerClass>
class ThreadManager: public ACE_Task<ACE_MT_SYNCH>, private IManager<SubWorkerClass>
{
public:

	typedef IWorker<SubWorkerClass> WorkerType;

	ThreadManager (ACE_UINT16 pool_size, ACE_INT16 max_timeout)
		: shutdown_(0), workers_lock_(), workers_cond_(workers_lock_),
		pool_size_(pool_size), max_timeout_(max_timeout)
	{
		ACE_TRACE ("Manager::Manager");
	}

	virtual int svc(void)
	{
		ACE_TRACE("Manager::svc");

		ACE_DEBUG((LM_INFO, ACE_TEXT("(%t) Manager started\n")));

		// Create pool.
		create_worker_pool();
		time_t btime = ACE_OS::time(0);
		while (!done())
		{
			ACE_Message_Block *mb = 0;
			ACE_Time_Value tv ((long)max_timeout_);
			tv += ACE_OS::time(0);
			ACE_Time_Value *ptv = max_timeout_ != -1 ? &tv : NULL;

			// Get a message request.
			if (this->getq(mb, ptv) < 0)
			{
				time_t etime = ACE_OS::time(0);
				ACE_DEBUG((LM_DEBUG, ACE_TEXT("process use time %d seconds.\n"), etime - btime));

				shut_down ();
				break;
			}

			// Choose a worker.
			WorkerType *worker = 0;
			{
				ACE_GUARD_RETURN (ACE_Thread_Mutex,
					worker_mon, this->workers_lock_, -1);

				while (this->workers_.is_empty())
					workers_cond_.wait();

				this->workers_.dequeue_head(worker);
			}

			// Ask the worker to do the job.
			worker->putq(mb);
		}

		return 0;
	}

	int shut_down(void)
	{
		ACE_TRACE ("Manager::shut_down");
		ACE_Unbounded_Queue<WorkerType*>::ITERATOR iter = this->workers_.begin ();
		WorkerType **worker_ptr = 0;
		do
		{
			iter.next (worker_ptr);
			WorkerType *worker = (*worker_ptr);
			ACE_Thread_ID id = thread_id(worker);
			char buf[65];
			id.to_string(buf);
			ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Attempting shutdown of %C\n"), buf));

			// Send the hangup message.
			ACE_Message_Block *mb = 0;
			ACE_NEW_RETURN(mb, ACE_Message_Block(0, ACE_Message_Block::MB_HANGUP), -1);
			worker->putq(mb);

			// Wait for the exit.
			worker->wait();

			ACE_ASSERT(worker->msg_queue()->is_empty());
			ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Worker %C shut down.\n"), buf));
			delete worker;
		}
		while (iter.advance());

		shutdown_ = 1;

		return 0;
	}

	const ACE_Thread_ID& thread_id(WorkerType *worker) { return worker->thread_id (); }

	virtual int return_to_work(WorkerType *worker)
	{
		ACE_GUARD_RETURN(ACE_Thread_Mutex,
			worker_mon, this->workers_lock_, -1);
		ACE_DEBUG((LM_DEBUG,
			ACE_TEXT("(%t) Worker %t returning to work.\n")));
		this->workers_.enqueue_tail(worker);
		this->workers_cond_.signal();

		return 0;
	}

private:

	int create_worker_pool(void)
	{
		ACE_GUARD_RETURN(ACE_Thread_Mutex,
			worker_mon,
			this->workers_lock_,
			-1);
		for (int i = 0; i < pool_size_; i++)
		{
			WorkerType *worker = 0;
			ACE_NEW_RETURN(worker, SubWorkerClass(this), -1);
			this->workers_.enqueue_tail(worker);
			worker->activate ();
		}

		return 0;
	}

	int done (void) { return (shutdown_ == 1); }

private:

	int shutdown_;
	ACE_Thread_Mutex workers_lock_;
	ACE_Condition<ACE_Thread_Mutex> workers_cond_;
	ACE_Unbounded_Queue<WorkerType*> workers_;
	ACE_UINT16 pool_size_;
	ACE_INT16 max_timeout_;
};


#endif //THREAD_MANAGER_H_